package com.afdor.rws.event.driven.channel.rocketmq;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.core.strategy.IdGeneratorStrategy;
import com.afdor.rws.event.driven.channel.ChannelProvider;
import com.afdor.rws.event.driven.channel.DefaultChannelProvider;
import com.afdor.rws.event.driven.channel.producer.EventProducer;
import com.afdor.rws.event.driven.channel.rocketmq.transaction.impl.RocketMqTransactionItemImpl;
import com.afdor.rws.event.driven.channel.transaction.ChannelTransactionItem;
import com.afdor.rws.event.driven.config.EventProducerConfig;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicMarkableReference;

/**
 * RocketMQ事件执行通道实现
 *
 * @author 悭梵
 * @date Created in 2018-12-18 11:33
 */
@NoArgsConstructor
@Extension(value = "rocketMqTransactionTopic", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = ChannelProvider.class)
public class RocketMqTransactionChannelProvider extends DefaultChannelProvider implements TransactionListener {

    /**
     * 生产者
     */
    private AtomicMarkableReference<EventProducer> eventProducerReference = new AtomicMarkableReference<>(null, false);

    /**
     * 采用内存管理事务项的方式不可靠，可靠方式应改为数据库或Redis等其他可靠存储方式来实现
     */
    private ConcurrentMap<String, RocketMqTransactionItem> localTransactionMap = new ConcurrentHashMap<>();
    private ConcurrentMap<String, RocketMqTransactionItem> transactionEventMap = new ConcurrentHashMap<>();

    @Override
    protected ChannelTransactionItem createChannelTransactionItem() {
        return new RocketMqTransactionItemImpl(String.valueOf(ExtensionLoader
                .getInstance(IdGeneratorStrategy.class)
                .getExtensionOrDefault(getConfig().getTransactionIdStrategyName())
                .generate()), this, transactionEventMap);
    }

    @Override
    protected EventProducer getEventProducer(EventProducerConfig eventProducerConfig) {
        if (!eventProducerReference.isMarked()) {
            EventProducer eventProducer = ExtensionLoader
                    .getInstance(EventProducer.class)
                    .newExtension(getConfig().getEventProducerStrategyName(), this)
                    .setEventSourceName(getEventSourceName())
                    .setSerialization(getSerialization())
                    .setConfig(eventProducerConfig)
                    .init();

            eventProducerReference.compareAndSet(null, eventProducer, false, true);
        }
        return eventProducerReference.getReference();
    }

    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        String eventId = o.toString();
        RocketMqTransactionItem rocketMqTransactionItem = new RocketMqTransactionItem(System.currentTimeMillis(), eventId, message.getTransactionId());
        transactionEventMap.put(eventId, rocketMqTransactionItem);
        localTransactionMap.put(message.getTransactionId(), rocketMqTransactionItem);
        return LocalTransactionState.UNKNOW;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        RocketMqTransactionItem transactionItem = localTransactionMap.get(messageExt.getTransactionId());
        LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
        if (null != transactionItem) {
            localTransactionState = transactionItem.getLocalTransactionState();
            if (Objects.equals(LocalTransactionState.UNKNOW, localTransactionState)) {
                if (System.currentTimeMillis() <= transactionItem.getStartTime() + 10 * 60 * 1000) {
                    localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
        }
        if (Objects.isNull(localTransactionState)) {
            localTransactionState = LocalTransactionState.UNKNOW;
        }
        if (!Objects.equals(LocalTransactionState.UNKNOW, localTransactionState)) {
            transactionEventMap.remove(transactionItem.getEventId());
            localTransactionMap.remove(transactionItem.getTransactionId());
        }
        return localTransactionState;
    }

    @Data
    public final class RocketMqTransactionItem {

        private final long startTime;
        private final String eventId;
        private final String transactionId;
        @Setter
        private volatile LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;

        public RocketMqTransactionItem(long startTime, String eventId, String transactionId) {
            this.eventId = eventId;
            this.startTime = startTime;
            this.transactionId = transactionId;
        }
    }

}
